package com.zhss.dfs.server;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhss.dfs.server.model.FileReplicaTask;
import com.zhss.dfs.server.model.ReplicaFile;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * 这个组件，就是负责管理集群里的所有的datanode的
 * @author zhonghuashishan
 *
 */
public class  DataNodeManager {

	private FSNamesystem fsNamesystem;

	public DataNodeManager(){
		// 启动就开始监控问题线程
		new DataNodeHealthMonitor().start();
	}

	/**
	 * 内存中维护的datanode列表
	 */
	private Map<String, DataNodeInfo> dataNodes = new ConcurrentHashMap<String, DataNodeInfo>();

	public FSNamesystem getFsNamesystem() {
		return fsNamesystem;
	}

	public void setFsNamesystem(FSNamesystem fsNamesystem) {
		this.fsNamesystem = fsNamesystem;
	}

	public Map<String, DataNodeInfo> getDataNodes() {
		return dataNodes;
	}

	public void setDataNodes(Map<String, DataNodeInfo> dataNodes) {
		this.dataNodes = dataNodes;
	}

	/**
	 * datanode进行注册
	 * @param ip 
	 * @param hostname
	 */
	public Boolean register(String ip, String hostname, Integer port) {
		DataNodeInfo datanode = new DataNodeInfo(ip, hostname, port);
		if (dataNodes.containsKey(ip + "-" + hostname)) {
			return false;
		}
		dataNodes.put(ip + "-" + hostname, datanode);
		return true;
	}


	/**
	 *  进行心跳检测
	 * @param ip
	 * @param hostname
	 * @return
	 */
	public Boolean heartBeat(String ip, String hostname) {
		DataNodeInfo dataNodeInfo = dataNodes.get(ip + "-" + hostname);
		if (dataNodeInfo == null) {
			// 存在说明nameNode重启后已经将dataNode移除，此时需要dataNode重新进行注册，上报
			return false;
		}
		dataNodeInfo.setLatestHeatBeatTime(System.currentTimeMillis());
		return true;
	}

	public String getAllocateDataNode(long fileSize) {
		List<DataNodeInfo> dataNodes = this.dataNodes.values().stream()
				.sorted(Comparator.comparing(DataNodeInfo::getFileStoreSize))
				.limit(2)
				.collect(Collectors.toList());
		// 默认文件已经存储在当前dataNode上了
		dataNodes.forEach(dataNodeInfo -> dataNodeInfo.addFileStoreSize(fileSize));
		return JSONObject.toJSONString(dataNodes);
	}

	public DataNodeInfo getDataNode(String hostName, String ip) {
		return dataNodes.get(ip + "-" + hostName);
	}

	/**
	 *  监控dataNode是否健康
	 */
	class DataNodeHealthMonitor extends Thread{

		@Override
		public void run() {
			try{
				while (true) {
					ArrayList<String> unHealthDataNodes = new ArrayList<String>();
					for (DataNodeInfo dataNodeInfo : dataNodes.values()) {
						if (System.currentTimeMillis() - dataNodeInfo.getLatestHeatBeatTime() > 90 * 1000) {
							//当前dataNode存在问题
							unHealthDataNodes.add(dataNodeInfo.getIp() + "-" + dataNodeInfo.getHostname());
						}
					}
					if (!unHealthDataNodes.isEmpty()) {

						for (String unHealthDataNode : unHealthDataNodes) {
							// 不健康节点进行移除
							dataNodes.remove(unHealthDataNode);
							// 移除副本信息
							removeDataNodeReplicaInfo(unHealthDataNode);
						}

					}
					// 每隔30秒检测一次
					Thread.sleep(30 * 1000);
				}
			} catch (Exception e){
				e.printStackTrace();
			}
		}

		private void removeDataNodeReplicaInfo(String unHealthDataNode) {
			String deadHostname = unHealthDataNode.split("-")[0];
			List<ReplicaFile> replicaFileList = fsNamesystem.getFileReplica(deadHostname);
			for (ReplicaFile replicaFile : replicaFileList) {
				DataNodeInfo sourceDataNode = fsNamesystem.getSourceDataNode(replicaFile.getFileName(), deadHostname);
				DataNodeInfo destDataNode = allocateDateNode(replicaFile.getFileSize());
				if (destDataNode != null && sourceDataNode != null) {
					FileReplicaTask fileReplicaTask = new FileReplicaTask(replicaFile.getFileSize(),replicaFile.getFileName(),sourceDataNode,destDataNode);
					// 将复制任务加入到复制任务队列中
					destDataNode.addReplaceTask(fileReplicaTask);
				}
			}
		}

		private DataNodeInfo allocateDateNode(Long fileSize) {
			synchronized (this) {
				DataNodeInfo dataNodeInfo = dataNodes.values().stream()
						.sorted(Comparator.comparing(DataNodeInfo::getFileStoreSize))
						.limit(1)
						.findFirst().orElse(null);
				if (dataNodeInfo != null) {
					dataNodeInfo.addFileStoreSize(dataNodeInfo.getFileStoreSize() + fileSize);
				}
				return dataNodeInfo;
			}
		}
	}
}
